/**
 * Created by emily on 17-11-15.
 */
import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.util.Bytes;


public class HbaseToKafka {

    public void hbaseToKafka() throws IOException {
        HbaseBasic hb = new HbaseBasic();
        KafkaBasic kb = new KafkaBasic();
        try{
            ArrayList<Result> res = hb.getResultScann("BridgeRealTime",
                    "5AB001-DX_1504228000", "5AB001-DX_1504228010660");
            ArrayList<Messg> messgs = new ArrayList<Messg>();



            for(Result rs: res){
                Long basetime = Long.valueOf(Bytes.toString(rs.getRow()).split("_")[1]);
                Cell[] cells = rs.rawCells();

                for(int i = 0;i<cells.length;i++){
                    short offset =  Bytes.toShort(cells[i].getQualifierArray(),cells[i].getQualifierOffset());
                    Long time = basetime + offset;
                    double value = Bytes.toDouble(cells[i].getValueArray(),cells[i].getValueOffset());

                    if(value<0){
                        value = 0;
                    }
                    System.out.println(time + "," + value);




                    Messg messg = new Messg("5AB001-DX", String.valueOf(time), String.valueOf(value));
                    messgs.add(messg);
                }
        }

        //kb.sendMessage("5AB001-DX", messgs );
            System.out.print("hello");

    }finally {

        }
    }

    public static void main(String[] args) throws IOException{
        HbaseToKafka hk = new HbaseToKafka();
        hk.hbaseToKafka();


    }
}
